package part02

import org.apache.spark.{SparkConf, SparkContext}

import java.util.regex.Pattern

object LogAnalysis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init)
      .setMaster("local[*]")

    val sc = new SparkContext(conf)
    //日志格式：IP	命中率(Hit/Miss)	响应时间	请求时间	请求方法	请求URL	请求协议	状态码	响应大小referer 用户代理
    val rdd1 = sc.textFile("data/cdn.txt")

    val rdd1Cache = rdd1.map(_.split("\\s+"))
    val rdd1Cache1 = rdd1Cache.cache()

    //    2.1、计算独立IP数
    val count = rdd1Cache1
      .map(_ (0))
      .distinct()
      .count()
    println(s"独立IP数是：$count")
    //2.2、统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）
    val ipPattern: Pattern = Pattern.compile("""(\S+) .+/(\S+\.mp4) .*""")
    rdd1.map(line => {
      val matchF = ipPattern.matcher(line)
      if (matchF.matches()) {
        (matchF.group(2), matchF.group(1))
      } else {
        ("", "")
      }
    }).distinct()
      .filter(f => {
        f._1 != "" && f._1 != ""
      })
      .map(f => (f._1, 1))
      .reduceByKey(_ + _)
      .sortBy(_._2)
      .collect.foreach(println)

    //2.3、统计一天中每个小时的流量
    rdd1Cache1.map(f => (f(3).split("\\:")(1), f(8), f(9)))
      .filter(f => f._2.contains("200") || f._2.contains("304") || f._2.contains("206"))
      .map(f => (f._1, f._3.toLong))
      .reduceByKey(_ + _)
      .mapValues(_ / 1024.0 / 1024 / 1024)
      .sortBy(_._1)
      .collect.foreach(println)

    sc.stop()
  }
}
